package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * com.atguigu.realtime.app.dwd.db.Dwd_10_DwdTradeRefundPaySuc
 * 退款
 */
public class Dwd_11_DwdToolCouponGet extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_11_DwdToolCouponGet().init(
                3011,
                2,
                "Dwd_11_DwdToolCouponGet"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        readOdsDb(tEnv, "Dwd_11_DwdToolCouponGet");

        // TODO 4. 读取优惠券领用数据，封装为表
        Table result = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['coupon_id'] coupon_id,\n" +
                "data['user_id'] user_id,\n" +
                "date_format(data['get_time'],'yyyy-MM-dd') date_id,\n" +
                "data['get_time'] get_time,\n" +
                "ts \n" +
                "from ods_db \n" +
                "where `database`='gmall2022' \n" +
                "and `table` = 'coupon_use'\n" +
                "and `type` = 'insert'\n");
        tEnv.createTemporaryView("coupon_use", result);
        //tEnv.sqlQuery("select * from  coupon_use").execute().print();
        //tEnv.sqlQuery("select * from " + couponUse).execute().print();


        // TODO 5. 建立 Kafka-Connector dwd_tool_coupon_get 表
        tEnv.executeSql("create table dwd_tool_coupon_get (\n" +
                "id string,\n" +
                "coupon_id string,\n" +
                "user_id string,\n" +
                "date_id string,\n" +
                "get_time string,\n" +
                "ts string\n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TOOL_COUPON_GET));

        // TODO 6. 将数据写入 Kafka-Connector 表
        result.executeInsert("dwd_tool_coupon_get");

        //tEnv.executeSql("insert into dwd_tool_coupon_get select * from result_table");

    }
}
